package com.gitee.huanminabc.jcommon.multithreading;

import com.gitee.huanminabc.jcommon.dsa.TimeIntervalKey;
import com.gitee.huanminabc.jcommon.multithreading.executor.ExecutorUtil;
import com.gitee.huanminabc.jcommon.multithreading.executor.SleepTools;
import lombok.Setter;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * 同一个 key 在时间区间内只能一个线程执行,其余的会等待前一个执行完毕在执行
 * key 必须实现hashCode方法
 * @Author huanmin
 * @Date 2024/6/6 下午4:38
 */
public class SimpleLockByTimeIntervalKey<K > {
     //时间区间,单位毫秒
    @Setter
    private  int timeInterval;

    private final Map<Integer, Queue<Runnable>> lockMap = new ConcurrentHashMap<>();
    private final Map<Integer,Long> timeClean = new ConcurrentHashMap<>();
    private final Set<Integer> concurrentSet ;
    private final Set<Integer> concurrentFun ;
    private   Function<K,Integer> hashCodeFunction;

    //此构造函数的key必须是字符串,否则自动获取对象的toString方法作为key
    public SimpleLockByTimeIntervalKey(int timeInterval) {
        if (timeInterval < 1) {
            throw new IllegalArgumentException("时间区间不能小于1毫秒,否则意义不大");
        }
        concurrentSet=new ConcurrentHashSet<>();
        concurrentFun=new ConcurrentHashSet<>();//判断是否函数执行
        this.timeInterval = timeInterval;

        //默认的hashCodeFunction
        this.hashCodeFunction = (key)-> TimeIntervalKey.decorateKey(key.toString(), timeInterval);
        //创建定时器,每隔一段时间清除队列是空的key
        new Thread(()->{
            while (true){
                SleepTools.second(1);
                timeClean.forEach((key, value) -> {
                    if (System.currentTimeMillis() > value){
                       lockMap.remove(key);
                        concurrentSet.remove(key);
                        concurrentFun.remove(key);
                        timeClean.remove(key);
                    }
                });
            }
        }).start();
    }

    //timeInterval 必须有的, 如果不想那么可以指定一个10ms的时间区间或者10~1000ms的时间区间
    //不能享受时间区间自动解锁锁, 因为是自定义的hashCode生成方式, 但是能保证key过期时间内只能执行一次
    public SimpleLockByTimeIntervalKey(int timeInterval ,Function<K,Integer> hashCodeFunction) {
        this(timeInterval);
        this.hashCodeFunction = hashCodeFunction;
    }
    //不能享受时间区间自动解锁锁, 因为是自定义的hashCode生成方式, 但是能保证key过期时间内只能执行一次
    public SimpleLockByTimeIntervalKey(Function<K,Integer> hashCodeFunction) {
        this(100,hashCodeFunction);
    }

    public void lock(K key, Runnable runnable) {
        lock(key,false,runnable);
    }
    public void lock(K key,boolean discard, Runnable runnable) {
        lock(key,timeInterval,discard,runnable);
    }
    /**
     * 给key加时间区间锁,到期自动解锁,同一个key在时间区间内只能一个线程执行其他的会等待前一个执行完毕
     * discard 丢弃并发的key ,当key在时间区间内有多个线程执行,那么丢弃后面的key
     */
    public void lock(K key,int timeInterval__,boolean discard, Runnable runnable) {
        int hashcode = hashCodeFunction.apply(key);
        //锁的结束时间
        long endTime = System.currentTimeMillis() + timeInterval__;
        //判断是否有锁,如果有锁,那么将任务加入到队列中
        if ( !concurrentSet.add(hashcode)) {
            if(discard){ //丢弃并发的key,不用担心清理的问题,因为有全局的清理线程
              return;
            }
            //给时间区间内的key加锁
            Queue<Runnable> blockingQueue = lockMap.get(hashcode);
            //如果是空的,那么等待,但是最多等待时间是时间区间
            while (blockingQueue == null&&System.currentTimeMillis()<endTime) {
                blockingQueue = lockMap.get(hashcode);
                SleepTools.ms(10);//防止频率过高
            }
            //如果在等待时间内没有获取到锁,那么换到下一个时间执行这个任务
            if (blockingQueue==null){
                lock(key,discard,runnable);
                return;
            }
            //只有第一次加入队列的时候才会创建一个线程
            if (concurrentFun.add(hashcode)){
                //创建一个线程,一直执行到时间区间结束,过程中持续执行队列中的任务
                Queue<Runnable> finalBlockingQueue = blockingQueue;
                ExecutorUtil.create(()->{
                    //一直执行到时间区间结束
                    while (System.currentTimeMillis() < endTime) {
                        Runnable poll = finalBlockingQueue.poll();

                        if (poll != null) {
                            poll.run();
                        }else{
                            synchronized (finalBlockingQueue){
                                try {
                                    finalBlockingQueue.wait(timeInterval__);//睡眠
                                } catch (InterruptedException ignored) {
                                }
                            }
                        }

                    }
                    //执行剩余的任务
                    while (!finalBlockingQueue.isEmpty()) {
                        Runnable poll = finalBlockingQueue.poll();
                        poll.run();
                    }
                    //时间区间结束,清除key队列
                    lockMap.remove(hashcode);
                    //清除key锁
                    concurrentSet.remove(hashcode);
                    //清除函数锁
                    concurrentFun.remove(hashcode);
                });
            }

            //唤醒
            synchronized (blockingQueue){
                blockingQueue.add(runnable);
                blockingQueue.notify();
            }
            return;
        }
        runnable.run();
        //没有锁,那么创建一个队列,并且加入到map中
        Queue<Runnable> blockingQueue = new LinkedBlockingQueue<>();
        lockMap.put(hashcode, blockingQueue);
        timeClean.put(hashcode,endTime);
    }

    //返回值不能是null否则无限等待
    public <V> V lock(K key, Supplier<V> supplier) {
        AtomicReference<V> result = new AtomicReference<>();
        lock(key, () -> {
            result.set(supplier.get());
        });
        //持续等待
        while (result.get() == null) {
            SleepTools.ms(10);
        }
        return result.get();
    }



    //丢弃并发的key, 后面的返回的全是null
    public <V> V discard(K key, Supplier<V> supplier) {
        AtomicReference<V> result = new AtomicReference<>();
        lock(key, true, () -> result.set(supplier.get()));
        return result.get();
    }

    public <V> V discard(K key,int timeInterval__, Supplier<V> supplier) {
        AtomicReference<V> result = new AtomicReference<>();
        lock(key,timeInterval__,true, () -> result.set(supplier.get()));
        return result.get();
    }

    public void discard(K key, Runnable  runnable) {
        lock(key, true, runnable);
    }
    public void discard(K key,int timeInterval__, Runnable  runnable) {
        lock(key,timeInterval__,true, runnable);
    }


    //等待执行完毕,避免主线程提前结束
    public void waitAll() {
        while (!lockMap.isEmpty()) {
            SleepTools.ms(10);
        }
    }
}
